package PrjHttpWs

/**
  使用json格式传递
    适合与H5进行交互
*/

import (
	"dgo/goutils/utils"
	"gitee.com/ymofen/supervalue"
	"net/http"
	"sync/atomic"

	"golang.org/x/net/websocket"
)

type WsHttpService struct {
	ws_req_url          string
	online              int32
	LogTypeStr          string
	handefun            func(pattern string, handler http.Handler)
	unhandlefun         func(pattern string)
	TransAttachInfoFlag bool // 转发时添加附加信息, 全局开启, 所有通道都可以收到, 也可以客户端订阅时,单独设定
	sc                  *utils.SubscribeCenter
	monitor_SendSize    *utils.MonitorSpeedSizeRec
	monitor_RecvSize    *utils.MonitorSpeedSizeRec
	OnWsConnected       func(sender *WsHttpService, conn *WsSvrConn) bool
	OnTransTopic        func(sender *WsHttpService, fromconn *WsSvrConn, topic string, rcv *wrapper.SuperValue) (data interface{})

	OnTransTopicEx func(sender *WsHttpService, fromconn *WsSvrConn, topic string, rcv *wrapper.SuperValue)
}

func NewWsHttpServiceWithMux(ws_req_url string, mux *http.ServeMux) *WsHttpService {
	rval := &WsHttpService{
		sc: utils.NewSubscribeCenterASync(),
	}

	rval.sc.ConfigASyncTaskDefaultMaxQueue(10240)
	mux.Handle(ws_req_url, websocket.Handler(rval.onHttpHandle))
	rval.sc.SubscribeChannel(WS_HTTP_CHANNEL_CMD, rval, rval.onRecvCmd)
	return rval
}

func NewWsHttpServiceHandleEx(handefun func(pattern string, handler http.Handler), unhandlefun func(pattern string)) *WsHttpService {
	rval := &WsHttpService{
		sc:          utils.NewSubscribeCenterASync(),
		handefun:    handefun,
		unhandlefun: unhandlefun,
	}

	rval.sc.ConfigASyncTaskDefaultMaxQueue(10240)
	rval.sc.SubscribeChannel(WS_HTTP_CHANNEL_CMD, rval, rval.onRecvCmd)
	return rval
}

func NewWsHttpServiceHandleFunc(ws_req_url string, handefun func(pattern string, handler http.Handler)) *WsHttpService {
	rval := &WsHttpService{
		sc:         utils.NewSubscribeCenterASync(),
		ws_req_url: ws_req_url,
	}

	rval.sc.ConfigASyncTaskDefaultMaxQueue(10240)

	handefun(ws_req_url, websocket.Handler(rval.onHttpHandle))
	rval.sc.SubscribeChannel(WS_HTTP_CHANNEL_CMD, rval, rval.onRecvCmd)
	return rval
}

func NewWsHttpService(ws_req_url string) *WsHttpService {
	return NewWsHttpServiceWithMux(ws_req_url, http.DefaultServeMux)
}

func (this *WsHttpService) Online() int32 {
	return this.online
}

func (this *WsHttpService) ServiceURL() string {
	return this.ws_req_url
}

func (this *WsHttpService) ExStatus() string {
	return this.sc.StatusString()
}

func (this *WsHttpService) ConfigMonitor(monitor_send, monitor_recv *utils.MonitorSpeedSizeRec) {
	this.monitor_RecvSize = monitor_recv
	this.monitor_SendSize = monitor_send
}

func (this *WsHttpService) DispatchMessage(topic string, data interface{}) {
	if str, ok := data.(string); ok {
		var jB utils.JSONBuilder
		jB.ObjectBegin()
		jB.AddEscapeStr(WS_TOPIC_KEY, topic).Spliter()
		jB.AddEscapeStr(WS_MSG_KEY, str)
		jB.ObjectEnd()
		this.sc.DispatchData(topic, nil, jB.String(), nil)
	} else if json, ok := data.(*wrapper.SuperValue); ok {
		json.SetKeyValue(WS_TOPIC_KEY, topic)
		this.sc.DispatchData(topic, nil, json.String(), nil)
	} else {
		json := wrapper.NewSVObject()
		json.SetKeyValues(WS_TOPIC_KEY, topic, WS_MSG_KEY, data)
		this.sc.DispatchData(topic, nil, json.String(), nil)
	}
}

func (this *WsHttpService) Close() {
	if this.unhandlefun != nil && len(this.ws_req_url) > 0 {
		this.unhandlefun(this.ws_req_url)
	}
}

func (this *WsHttpService) ConfigServiceUrl(url string) (changed bool) {
	if this.ws_req_url != url {
		if this.unhandlefun != nil {
			this.unhandlefun(this.ws_req_url)
		}
		this.ws_req_url = url

		if len(this.ws_req_url) > 0 {
			if this.handefun != nil {
				this.handefun(this.ws_req_url, websocket.Handler(this.onHttpHandle))
			}
		}
		return true
	}
	return false
}

func (this *WsHttpService) onRecvCmd(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	if sender == nil {
		utils.Log.WarnTypef(this.LogTypeStr, "命令没有Sender")
		return true
	}
	wsconn := sender.(*WsSvrConn)
	if json, ok := data.(*wrapper.SuperValue); ok {
		cmd := json.StringByName("cmd", "")
		if cmd == WS_HTTP_CMD_subscribe {
			str := json.StringByName("channels", "")
			r := wsconn.SubscribleChannel(str)
			json.CheckExistsIntFunc("request-attach", func(val int64) {
				wsconn.trans_attach_info = val == 1
			})

			wsconn.SendKVS("code", 0, "channels", str, "n", r, "client-addr", wsconn.remoteAddr)
		} else if cmd == WS_HTTP_CMD_unsubscribe {
			str := json.StringByName("channels", "")
			r := wsconn.UnsubscribleChannel(str)
			wsconn.SendKVS("code", 0, "channels", str, "n", r)
		}
	}
	return true
}

func (this *WsHttpService) onRecv(wsconn *WsSvrConn, msgtype byte, msg []byte) {
	var strHex string
	defer utils.PanicHandlerWithDebugString(strHex)

	if msgtype == WS_MESSAGE_TEXT {
		//if !utils.IsUtf8Buf(msg, 128){
		//	var strmsg string
		//	if len(msg) < 128{
		//		strmsg = string(msg)
		//	}else{
		//		strmsg = string(msg[:127])
		//	}
		//	utils.Log.WarnTypef(this.LogTypeStr, "非法的UTF8数据, 来自:%s, 忽略处理, :%s", wsconn.remoteAddr, strmsg)
		//	return
		//}

		strHex = utils.BufToHexStr(msg, 0, "")

		json, err := wrapper.NewSVFromBuf(msg)
		if err != nil {
			var strmsg string
			if len(msg) < 128 {
				strmsg = string(msg)
			} else {
				strmsg = string(msg[:127]) + "..."
			}
			utils.Log.WarnTypef(this.LogTypeStr, "非法的数据:%s, from:%s, err:%s", strmsg, wsconn.remoteAddr, err.Error())
			return
		}
		topic := json.StringByName(WS_TOPIC_KEY, "")
		if len(topic) == 0 {
			topic = "unkown"
		}

		if topic == "__ECHO" {
			wsconn.SendV(json)
			return
		}

		if this.TransAttachInfoFlag {
			info := json.ForceByPath("__clients")
			info.SetKeyValue("from-addr", wsconn.remoteAddr)
		}

		if this.OnTransTopicEx != nil {
			this.OnTransTopicEx(this, wsconn, topic, json)
		} else {
			if this.OnTransTopic != nil {
				rcv := this.OnTransTopic(this, wsconn, topic, json)
				if rcv != nil {
					this.sc.DispatchData(topic, wsconn, rcv, nil)
				} else {
					this.sc.DispatchData(topic, wsconn, json, nil)
				}
			} else {
				this.sc.DispatchData(topic, wsconn, json, nil)
			}
		}
	} else {

	}
}

func (this *WsHttpService) onHttpHandle(ws *websocket.Conn) {
	defer utils.PanicHandlerWithDebugString("WsHttpService-onHttpHandle")
	wsconn := NewWsSvrConn(ws, this)
	atomic.AddInt32(&this.online, 1)
	defer func() {
		wsconn.onClose()
		atomic.AddInt32(&this.online, -1)
	}()
	if this.OnWsConnected != nil {
		if !this.OnWsConnected(this, wsconn) {
			return
		}
	}
	for {
		msgtype, msg, err := ReceiveMessage(ws)
		if err != nil {
			utils.Log.DebugTypef(this.LogTypeStr, "接收消息出现异常:%s", err.Error())
			break
		}

		lvMonitor := this.monitor_RecvSize
		if lvMonitor != nil {
			lvMonitor.Delta(int64(len(msg)))
		}

		this.onRecv(wsconn, msgtype, msg)
	}

}
